package com.hmc.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.xerial.snappy.Snappy;

import java.util.Properties;

/**
 * 配置生产者吞吐量，应答
 */
public class CustomProducerParameters {
    public static void main(String[] args) {
        Properties properties = CustomProducer.getProperties();
        //配置缓冲区
        configBuffer(properties);
        //配置应答
        ack(properties);
        //创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //发送数据
        for (int i = 0 ; i <= 10 ; i++) {
            producer.send(new ProducerRecord<>("first", "hmc"+i));
        }
        //关闭资源
        producer.close();
    }

    private static void ack(Properties properties) {
        //应答级别acks：（默认为-1即all）leader以及irs列别中的队列都收到后应答
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        //未收到应答后，再次发送的重试次数(默认为int的最大值)
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
    }

    private static void configBuffer(Properties properties) {
        //batch.size大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16*1024);
        //linger.ms等待时间
        properties.put(ProducerConfig.LINGER_MS_CONFIG,50);
        //compression.type压缩算法
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        //RecordAccumulator缓冲区大小
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,32*1024*1024);
    }
}
